package com.els.connection;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Vector;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequestBuilder;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;

import com.bridge.domain.PropertyVO;
import com.els.domain.QueryRespVO;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;

public abstract class AbstractElastic {
	protected Logger logger=Logger.getLogger(this.getClass());
	protected Client client;
	protected String esUrl;
	protected String cluster;
	protected String shieldUser;
	protected int _buffer=5000;
	protected long _timeout_s=3000l;
	protected long _timeout_l=600000l;
	public Client getClient() {
		if(client==null){
			buildClient();
		}
		return client;
	}

	public void setClient(Client client) {
		this.client = client;
	}
	public abstract boolean buildClient();
	public void closeClient() {
		client.close();
		client=null;
	}
	
	public int get_buffer() {
		return _buffer;
	}

	public void set_buffer(int _buffer) {
		this._buffer = _buffer;
	}
	public long getTotal(String index,String type){
		SearchRequestBuilder srb;
		buildClient();
		if(type==null){
			srb= client.prepareSearch(index);
		}
		else {
			srb= client.prepareSearch(index).setTypes(type);
		}
//		srb.setSearchType(SearchType.COUNT);
		srb.setSize(0);
		SearchResponse sr;
		try{
			sr = srb.execute().actionGet();
			
		}catch(Exception e){
			logger.error("getTotal has error!");
			e.printStackTrace();
			return 0;
		}finally{
			closeClient();
		}

		long result= sr.getHits().getTotalHits();
		return result;
	}
	/**
	 * @param index 索引名
	 * @param type 类型名
	 * @param jsonList 导入数据列表，json格式
	 * @param id 导入的id命名取值，当为null时，则由ELS自动生成
	 * @return 返回true说明导入成功，返回false说明导入失败
	 */
	protected boolean bulkIndex(String index, String type,
			List<JsonObject> jsonList, String id) {
		List<IndexRequestBuilder> requestList=new Vector<IndexRequestBuilder>();
		IndexRequestBuilder request;
		for (int i=0;i<jsonList.size();i++) {
			logger.debug("idx-i:"+i);
			logger.debug(jsonList.get(i));
			request = prepareInsertOne(index, type, id, jsonList.get(i));
			requestList.add(request);
		}
		return bulk(requestList);
	}
	/**
	 * @param index 索引名
	 * @param type 类型名
	 * @param jsonList 导入数据列表，json格式
	 * @param id 导入的id命名取值，当为null时，则由ELS自动生成
	 * @return 返回true说明导入成功，返回false说明导入失败
	 */
	protected boolean bulkIndex(String index, String type,
			List<JsonObject> jsonList, String id,String parent) {
		List<IndexRequestBuilder> requestList=new Vector<IndexRequestBuilder>();
		for (int i=0;i<jsonList.size();i++) {
			 IndexRequestBuilder request = prepareInsertOne(index, type, id, parent, jsonList.get(i));
			 requestList.add(request);
		}
		return bulk(requestList);
	}
	protected <T> boolean bulk(List<T> requestList){
		buildClient();
		BulkRequestBuilder bulk = client.prepareBulk();
		for (T request : requestList) {
			if(request instanceof IndexRequestBuilder){
				bulk.add((IndexRequestBuilder) request);
			}else if(request instanceof DeleteRequestBuilder){
				bulk.add((DeleteRequestBuilder) request);
			}else if(request instanceof UpdateRequestBuilder){
				bulk.add((UpdateRequestBuilder) request);
			}
		}
		logger.info("Request Bulk begins");
		TimeValue timeout=new TimeValue(_timeout_l);
		BulkResponse bulkResponse;
		try{
			bulkResponse = bulk.setTimeout(timeout).execute().actionGet();
		}catch(Exception e){
			logger.error("bulk has failure!");
			e.printStackTrace();
			return false;
		}finally{
			closeClient();
		}

		logger.info("Request Bulk end");
		if (bulkResponse.hasFailures()) {
			String errorMessage=bulkResponse.buildFailureMessage();
			logger.error("bulk has failure!");
			logger.error(errorMessage);
			return false;
		}else {
			logger.info("bulk success");
			return true;
		}
	}

	public IndexRequestBuilder prepareInsertOne(String indexName, String typeName, String id,JsonObject obj) {
		String _id=get_Id(obj, id);
		buildClient();
		IndexRequestBuilder request = client.prepareIndex();
		request.setIndex(indexName).setType(typeName).setId(_id);
//		XContentBuilder xcb = getXContentBuilder(obj);
		if(obj!=null){
			request.setSource(obj.toString());
			TimeValue timeout=new TimeValue(_timeout_s);
			request.setTimeout(timeout);
			return request;
		}else{
			return null;
		}		
	}
	public IndexRequestBuilder prepareInsertOne(String indexName, String typeName, String id, String parent,JsonObject obj) {
		String _parent=null;
		try {
			_parent=obj.get(parent).getAsString();
		} catch (Exception e) {
			logger.error("wrong parent id, cannot find COLUMN["+parent+"] in JSONObject");
			return null;
		}
		
		String _id=get_Id(obj, id);
		buildClient();
		IndexRequestBuilder request = client.prepareIndex();
		request.setIndex(indexName).setType(typeName).setId(_id).setParent(_parent);
		XContentBuilder xcb = getXContentBuilder(obj);
		if(xcb!=null){
			request.setSource(xcb);
			TimeValue timeout=new TimeValue(_timeout_s);
			request.setTimeout(timeout);
			return request;
		}else{
			return null;
		}		
	}
	public String insertOne(String indexName, String typeName, String id,
			JsonObject obj) {
		IndexRequestBuilder request = prepareInsertOne(indexName, typeName, id, obj);
		if(request!=null){
			try{
				IndexResponse response = request.execute().actionGet();
				return response.getId();
			}catch(Exception e){
				logger.error("insertOne has an error!");
				e.printStackTrace();
				return null;
			}finally{
				closeClient();
			}
		}else{
			return null;
		}		
	}
	public DeleteRequestBuilder prepareDeleteOne(String indexName, String typeName,String _id){
		buildClient();
		DeleteRequestBuilder request = client.prepareDelete();
		request.setIndex(indexName).setType(typeName).setId(_id);
		TimeValue timeout=new TimeValue(_timeout_s);
		request.setTimeout(timeout);
		return request;
	}
	public boolean deleteOne(String indexName, String typeName,String _id){
		buildClient();
		DeleteRequestBuilder request = prepareDeleteOne(indexName, typeName, _id);
		try {
			DeleteResponse response = request.execute().actionGet();
			return response.isFound();
		} catch (Exception e) {
			logger.error("deleteOne has an error!");
			e.printStackTrace();
			return false;
		}finally{
			closeClient();
		}
	}
	public UpdateRequestBuilder prepareUpdateOne(String indexName, String typeName,String _id, JsonObject obj){
		buildClient();
		UpdateRequestBuilder request = client.prepareUpdate();
		request.setIndex(indexName).setType(typeName).setId(_id);
		XContentBuilder source=getXContentBuilder(obj);
		try {
			request.setDoc(source);
		} catch (Exception e) {
			logger.error("update failure! JSONObject don't fit XContentBuilder!");
			return null;
		}
		TimeValue timeout=new TimeValue(_timeout_s);
		request.setTimeout(timeout);
		return request;
	}
	public boolean updateOne(String indexName, String typeName,String _id, JsonObject obj){
		UpdateRequestBuilder request = prepareUpdateOne(indexName, typeName, _id, obj);
		if(request==null) return false;
		try{
			request.execute().actionGet();
			return true;
		}catch(Exception e){
			logger.error("Do not find row to update!");
			return false;
		}finally{
			closeClient();
		}
	}
	public JsonObject selectOne(String indexName, String typeName,String id, JsonObject obj){
		buildClient();
		SearchRequestBuilder request = client.prepareSearch(indexName);
		request.setTypes(typeName);
		if(id!=null){
			QueryBuilder idQuery=QueryBuilders.termQuery("_id", id);
			request.setQuery(idQuery);
		}else if(obj!=null&&!obj.isJsonNull()){
			BoolQueryBuilder query=QueryBuilders.boolQuery();
			for (Entry<String, JsonElement> element : obj.entrySet()) {
				String key=element.getKey();
				String value=((JsonElement)element.getValue()).getAsString();
				query.must(QueryBuilders.termQuery(key, value));
			}
			request.setQuery(query);
		}else {
			return null;
		}
		SearchResponse response;
		try{
			response = request.execute().actionGet();
		}catch(Exception e){
			logger.error("selectOne has an error!");
			e.printStackTrace();
			return null;
		}finally{
			closeClient();
		}

		SearchHit[] hits = response.getHits().getHits();
		if(response.getHits().getTotalHits()>0){
			return getOneJsonObjectFrom(hits[0]);
		}else{
			return null;
		}
	}

	public List<JsonObject> queryAll(String indexName, String typeName){
		buildClient();
		SearchRequestBuilder request = client.prepareSearch(indexName);
		request.setTypes(typeName);
		request.setQuery(QueryBuilders.matchAllQuery());
		List<JsonObject> results=scrollAll(request);
		return results;
	}
	private List<JsonObject> scrollAll(SearchRequestBuilder request){
		List<JsonObject> results=new Vector<JsonObject>();
		TimeValue keepAlive=new TimeValue(6000l);
		Scroll scroll=new Scroll(keepAlive); 
		request.setScroll(scroll);
		SearchResponse response;
		try{ 
			response= request.execute().actionGet();
		}catch(Exception e){
			logger.error("scrollAll has an error!");
			e.printStackTrace();
			closeClient();
			return null;
		}
		String scrollId=response.getScrollId();
		String tmpId="";
		while (response.getHits().getHits().length>0&&!tmpId.equals(scrollId)) {
			tmpId=scrollId;
			for(SearchHit hit:response.getHits().getHits()){
				JsonObject jsonObject=getOneJsonObjectFrom(hit);
				results.add(jsonObject);
			}
			try{
				response=client.prepareSearchScroll(scrollId).setScroll(scroll).execute().actionGet();
				scrollId=response.getScrollId();
			}catch(Exception e){
				e.printStackTrace();
				break;
			}
		}
		closeClient();
		return results;
	}
	public List<JsonObject> queryAllByQueryBuilder(String indexName, String typeName,QueryBuilder q){
		buildClient();
		SearchRequestBuilder request = client.prepareSearch(indexName).setTypes(typeName);
		request.setQuery(q);
		request.addSort("_id", SortOrder.ASC);
		List<JsonObject> results=scrollAll(request);
		return results;
	}
	/**
	 * @param 根据自定义的QueryBuilder进行查询
	 * @return 结果为空时，返回size为0的List
	 */
	public List<JsonObject> queryByQueryBuilder(String indexName, String typeName,QueryBuilder q,int from,int size){
		buildClient();
		SearchRequestBuilder request = client.prepareSearch(indexName).setTypes(typeName);
		request.setQuery(q);
		request.setFrom(from);
		request.setSize(size);
//		request.addSort("_id", SortOrder.ASC);
		SearchResponse response;
		try{
			response = request.execute().actionGet();
		}catch(Exception e){
			logger.error("queryByQueryBuilder has an error!");
			e.printStackTrace();
			return null;
		}finally{
			closeClient();
		}

		List<JsonObject> results=getResultFromResponse(response);
		return results;
	}
	/**
	 * 准备查询语句组织
	 * @param indexName
	 * @param typeName
	 * @param queryBuilder
	 * @return SearchRequestBuilder
	 */
	public SearchRequestBuilder prepareSearch(String indexName, String typeName,QueryBuilder queryBuilder){
		buildClient();
		SearchRequestBuilder srb;
		if(typeName==null){
			srb=client.prepareSearch(indexName).setQuery(queryBuilder);
		}else{
			srb=client.prepareSearch(indexName).setTypes(typeName).setQuery(queryBuilder);
		}
		return srb;
	}
	/**
	 * 根据srb，运行返回结果
	 * @param srb
	 * @return
	 */
	public QueryRespVO executeQueryRequest(SearchRequestBuilder srb){
		QueryRespVO qrv=new QueryRespVO();
		SearchResponse response;
		try{
			response=srb.execute().actionGet();
		}catch(Exception e){
			e.printStackTrace();
			logger.error("executeQueryRequest has an error!");
			logger.error(e.getMessage());
			return null;
		}finally{
			closeClient();
		}
		List<JsonObject> results=getResultFromResponse(response);
		qrv.setData(results);
		qrv.setTotal(response.getHits().getTotalHits());
		qrv.setAggs(response.getAggregations());
		qrv.setResponse(new Gson().fromJson(response.toString(), JsonObject.class));
		return qrv;
	}
	protected List<JsonObject> getResultFromResponse(SearchResponse response){
		List<JsonObject> results=new Vector<JsonObject>();
		for(SearchHit hit:response.getHits().getHits()){
			JsonObject obj=getOneJsonObjectFrom(hit);
			results.add(obj);
		}
		return results;
	}
	private JsonObject getOneJsonObjectFrom(SearchHit hit){
		JsonObject obj=new Gson().fromJson(hit.getSourceAsString(), JsonObject.class);
		Map<String, Object> source = hit.getSource();
		obj.addProperty("_id", hit.getId());
		obj.addProperty("_type", hit.getType());
		for (String key : source.keySet()) {
			if(hit.getHighlightFields().get(key)!=null){
				Text[] highlight = hit.getHighlightFields().get(key).getFragments();
				String text="";
				for (Text hl : highlight) {
					text+=hl.toString();
				}
				obj.addProperty(key, text);
			}
		}
		return obj;
	}
	private XContentBuilder getXContentBuilder(JsonObject jsonObject){
		XContentBuilder xcb=null;
			try {
				xcb= XContentFactory.jsonBuilder().startObject();
			Set<Entry<String, JsonElement>> entrySet = jsonObject.entrySet();
			for (Entry<?, ?> entry : entrySet) {
				String key=entry.getKey().toString();
				JsonElement value=(JsonElement)entry.getValue();
				if(!value.isJsonNull()){
					logger.debug(key+":"+value.getAsString());
					xcb.field(key,value.getAsString());
				}else{
					xcb.field(key,"");
				}
			}
			xcb.endObject();
		}catch(IOException e){
			logger.error("create XContentBuilder object failure!");
		}
		
		return xcb;
	}
	private String get_Id(JsonObject jsonObject,String id){
		String _id="";
		if(id!=null){
			String[] indexes=id.split("\\|\\|");
			for (String idx : indexes) {
				try {
					_id+=jsonObject.get(idx)==null?idx:jsonObject.get(idx).getAsString();
				} catch (Exception e1) {
						// TODO Auto-generated catch block
					logger.debug("use sequence id");
					_id+=idx;
					continue;
				}
			}
			
		}else {
			_id=null;
		}
		return _id;
	}
	//增加type
	public boolean newType(String index,String type,List<PropertyVO> mappings){
		buildClient();
		try{
			XContentBuilder mapping=XContentFactory.jsonBuilder().startObject().startObject("properties");
			if(mappings.size()>0){
				for (PropertyVO field : mappings) {
					mapping = mapping.startObject(field.getName());
					if(field.getType().substring(0,4).equals("date")){
						String data = field.getType().substring(0,4);
						int len = field.getType().length();
						String format = field.getType().substring(5,len);
						mapping.field("type",data)
							.field("format",format);
					}else{
						mapping.field("type",field.getType());
					}
					//是否分词
					if (field.isPpl()) {
						//只创mapping
//						mapping.field("indexAnalyzer","ik")
//									  .field("searchAnalyzer","ik")
//									  .field("store","true");
						mapping.field("store","true");
					} else{
						//只创mapping
						mapping.field("index","not_analyzed")
									  .field("store","true");
					}
					//是否.raw
					if (field.isRaw()) {
						mapping.startObject("fields")
									.startObject("raw")
										.field("type",field.getType())
										.field("store","true")
										.field("index","not_analyzed")
									.endObject()
									.endObject();
					}
					mapping.endObject();
				}
			}
			mapping.endObject().endObject();
			logger.info(mapping.string());
			//在已有的索引下添加类型
			PutMappingRequest addMapping = Requests.putMappingRequest(index).type(type).source(mapping);
			client.admin().indices().putMapping(addMapping).actionGet();
			return true;
		}catch(Exception e){
			logger.error("new type has exception!");
			logger.error(e.getMessage());
			return false;
		}finally{
			closeClient();
		}
	}
	//增加index
	public boolean newIndex(String index,int shardSize,int replicaSize){
		buildClient();
		try {
			Settings settings = Settings.settingsBuilder()
					.put("number_of_shards", shardSize)
					.put("number_of_replicas", replicaSize).build();
			client.admin().indices().prepareCreate(index).setSettings(settings)
					.execute().actionGet();
			return true;
		} catch (Exception e) {
			logger.error("new index has exception!");
			logger.error(e.getMessage());
			return false;
		}finally{
			closeClient();
		}
	}
}
